package raft

import (
	"bytes"
	"context"
	"edata-hub/pkg/raft/services"
	"encoding/gob"
	"fmt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

// @Description: 处理来自其他Master的日志复制请求
// @receiver r
// @param ctx
// @param req
// @return resp
// @return err
func (r *Raft) OnHandleAppendEntries(ctx context.Context, req *services.AppendEntriesReq) (resp *services.AppendEntriesResp, err error) {
	r.Mutex.Lock()
	defer r.Mutex.Unlock()
	r.ResetElectionTime()
	resp.Success = false
	resp.Term = r.Term
	if r.Role == Candidate {
		r.Role = Follower
	}

	if req.Term > r.Term {
		r.Term = req.Term
		resp.Success = true
		return //此时不需要马上复制日志，待下次心跳就会进入下面日志复制的流程，所以直接返回
	}
	if req.Term < r.Term {
		//原本的集群由于不可抗拒的因素停止，并且需要手动重启时，先启动的不一定是原本的 LEADER
		//先启动如果是原本的 FOLLOWER，他会先变成 LEADER，并且该 LEADER 的 Term 可能会比最新的 Term 旧
		//此时应该让发送消息过来的 LEADER 直接变成 FOLLOWER
		return
	}

	//Leader 记录的该 Follower 的最新日志 Index 比真实最新记录 Index 还要大
	//将 Follower 当前的最新日志 Index 返回，Leader 刷新对该 Follower 的最新日志 Index 记录。
	if r.Log.lastLog().Index < req.LastLogIndex {
		resp.Conflict = true
		resp.XTerm = -1
		resp.XIndex = -1
		resp.LastLogIndex = r.Log.lastLog().Index
		return
	}

	//相同日志Index的情况下，Leader和Follower所记录的Term不同
	//此时Leader和Follower会协力找出具有相同Index和Term的日志位置，寻找的方法基于以下几点认识
	//TODO 没想明白
	//if r.Ls.at(req.LastLogIndex).Term != req.LastLogTerm {
	//	resp.Conflict = true
	//	xTerm := r.Ls.at(req.LastLogIndex).Term
	//	//找到该 Follower 日志所标记的上一个任期一个日志对应的 Index
	//	for xIndex := req.LastLogIndex; xIndex > 0; xIndex-- {
	//		if r.Ls.at(xIndex).Term != xTerm {
	//			resp.XIndex = xIndex + 1
	//			break
	//		}
	//	}
	//	resp.XTerm = xTerm
	//	return
	//}
	var logs Logs
	decoder := gob.NewDecoder(bytes.NewBuffer(req.Data))
	decoder.Decode(logs.Entries)
	//到此处，已经说明Leader的任期与Follower的任期是相同的
	//由于Entries中存储的是相同任期的数据，则判断Leader和Follower日志的任期是否相同
	if logs.Entries[0].Term != r.Log.Entries[0].Term {
		//TODO 将旧任期的日志持久化
	} else {
		for idx, entry := range logs.Entries {
			if entry.Index > r.Log.lastLog().Index {
				r.Log.append(logs.Entries[idx:]...)
				break
			}
		}
	}
	resp.LastLogIndex = r.Log.lastLog().Index
	resp.Success = true
	return
}

// @Description: 处理来自Candidate的投票请求。是一个远程调用方法
// @receiver r
// @param ctx
// @param req
// @return resp
// @return err
func (r *Raft) OnHandleVoteReq(ctx context.Context, req *services.VoteReq) (resp *services.VoteResp, err error) {
	r.Msg.Info("%s receive vote request from %s,start voting", r.Addr, req.Address)
	resp = &services.VoteResp{}
	resp.Address = r.Addr
	resp.Term = r.Term
	resp.VoteYN = false
	r.ResetElectionTime()
	if r.Term > req.Term {
		resp.VoteYN = false
	}
	if r.Term < req.Term {
		r.Term = req.Term
		r.Role = Follower
		r.VoteFor = "YES"
		//TODO 将日志持久化，即，转成二进制
	}
	upToDate := req.LastLogTerm > r.Log.lastLog().Term ||
		(req.LastLogTerm == r.Log.lastLog().Term && req.LastLogIndex >= r.Log.lastLog().Index)
	if (r.VoteFor == "YES" || r.VoteFor == req.Address) && upToDate {
		resp.VoteYN = true
		r.VoteFor = req.Address
		//TODO 将日志持久化
	}
	return
}

/*处理基于RPC客户端发起的命令，包含对初始命令的优化，命令的执行，以及返回命令执行的结果*/
func (r *Raft) OnHandleCommand(req *services.CommandReq, server services.RaftService_OnHandleCommandServer) error {
	oce := CommandEntry{
		Name:       req.Command,
		Params:     req.Params,
		ExecStatus: 0,
	}
	//调用优化器，对初始命令进行优化，得到多个子命令
	rces, _ := r.CommandOptimizer(oce)
	// TODO 创建本次命令执行的进度机制
	for _, rce := range rces {
		resp_start := &services.CommandResp{
			Status: 0,
			Msg:    fmt.Sprintf("command %s is running", rce.Name),
		}
		if start_err := server.Send(resp_start); start_err != nil {
			r.Msg.Error("can not send start response to command %s:%s", rce.Name, start_err.Error())
			break
		}
		resp_end := &services.CommandResp{
			Status: 0,
			Msg:    "",
		}
		method, ok := r.Commands[rce.Name]
		if !ok {
			resp_end.Status = -1
			resp_end.Msg = fmt.Sprintf("command %s is not existed", rce.Name)
		} else {
			ret := method(rce.Params)
			//TODO 此处暂时没有修改队列中命令体的状态
			//r.CommandQueues[id][index].ExecStatus = ret.Status
			resp_end.Status = ret.Status
			resp_end.Msg = ret.Msg
			resp_end.Results = fmt.Sprintf("%s", ret.Data)
			//当前命令执行失败的情况下，后续操作不再执行
		}
		if end_err := server.Send(resp_end); end_err != nil {
			r.Msg.Error("can not send response to command %s:%s", rce.Name, end_err.Error())
			break
		}
		if resp_end.Status == -1 {
			break
		}
	}
	//所有命令执行成功的情况下，客户端此时还在等待服务端传回消息，如果命令集执行过去出现status=-1的情况，通道会关闭。
	resp := &services.CommandResp{
		Status: 2,
	}
	if err := server.Send(resp); err != nil {
		r.Msg.Error("can not send response: %v", err)
	}
	return nil
}

// 创建GRPC客户端，客户端的连接地址
func (r *Raft) CreateGRPCClient(addr string) (services.RaftServiceClient, error) {
	conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		return nil, err
	}
	client := services.NewRaftServiceClient(conn)
	return client, nil
}
